package ch.jamme.impl;

import java.util.concurrent.*;

import org.jdom.*;

import ch.jamme.exception.*;
import ch.jamme.reference.*;

class XmlToObjectProducer extends XmlToObjectConvertor {

	private int parallelThreadsNumber;
	
	private ExecutorService executorService;
	private BlockingQueue<XmlMarshallableReference> queue = new LinkedBlockingQueue<XmlMarshallableReference>();
	
	private Element rootElement;

	XmlToObjectProducer(MarshallerEngine marshallerEngine, int parallelThreadsNumber) {
		super(marshallerEngine);
		this.parallelThreadsNumber = parallelThreadsNumber;
		executorService = Executors.newFixedThreadPool(parallelThreadsNumber);
	}
	
	@Override
	protected <T> T convertXmlToObject(Element element, Class<T> fieldClass) throws UnmarshallException {
		rootElement = element;
		Runnable consumer = new XmlToObjectConsumer(marshallerEngine, queue);
		for (int i = 0; i < parallelThreadsNumber; i++) executorService.execute(consumer);
		
		@SuppressWarnings("unchecked")
		T object = (T)toObject(element, null, fieldClass);

		try {
			// put as many poison pills on the queue as the number of the consumer threads to signal them that their work is finished
			for (int i = 0; i < parallelThreadsNumber; i++) queue.put(XmlMarshallableReference.createVoidReference());
			
			// wait until all the consumer threads have finished
			executorService.shutdown();
			executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
		return object;
	}
	
	@Override
	protected void populateObject(Element element, Object object) throws UnmarshallException {
		if (element == rootElement) super.populateObject(element, object);
		else try {
			queue.put(new XmlMarshallableReference(element, object));
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}
}
